package cn.getech.data.development.service.impl;


import cn.getech.data.development.constant.*;
import cn.getech.data.development.dto.EsDto;
import cn.getech.data.development.dto.HBaseDto;
import cn.getech.data.development.dto.KuduDto;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.entity.permission.*;
import cn.getech.data.development.entity.vo.ConfConnectDetail;
import cn.getech.data.development.mapper.AnalysisInfoMapper;
import cn.getech.data.development.mapper.ConfConnectMapper;
import cn.getech.data.development.mapper.ProcConnectMapper;
import cn.getech.data.development.mapper.permission.SysRoleDataPermissionMapper;
import cn.getech.data.development.mapper.permission.SysTenantDataPermissionMapper;
import cn.getech.data.development.model.dto.CollectionTaskUpdateDbDto;
import cn.getech.data.development.model.dto.ConfConnectDto;
import cn.getech.data.development.model.qo.ConfConnectQO;
import cn.getech.data.development.model.req.confconnect.TableInfoFieldRelationReq;
import cn.getech.data.development.model.req.confconnect.TableInfoFieldReq;
import cn.getech.data.development.model.res.confconnect.TableField;
import cn.getech.data.development.model.res.confconnect.TableInfoFieldRelationRes;
import cn.getech.data.development.model.res.confconnect.TableInfoFieldRes;
import cn.getech.data.development.model.vo.UserCreaterVO;
import cn.getech.data.development.service.ConfConnectService;
import cn.getech.data.development.service.ConnectFieldInfoService;
import cn.getech.data.development.service.SapConnectFieldService;
import cn.getech.data.development.service.TableInfoService;
import cn.getech.data.development.utils.*;
import cn.getech.data.development.utils.kafka.LocalKafkaUtil;
import cn.getech.data.development.utils.sap.SAPConnUtils;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.*;
import cn.getech.system.center.constant.DefConnectEnum;
import cn.getech.system.center.constant.DepartmentTypeEnum;
import cn.getech.system.center.entity.Department;
import cn.getech.system.center.entity.SysTclDepartment;
import cn.getech.system.center.model.dto.DepartmentMinDto;
import cn.getech.system.center.service.IDepartmentService;
import cn.getech.system.center.service.ISysTclDepartmentService;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.http.Method;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Maps;
import com.jcraft.jsch.*;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.sap.conn.jco.JCoDestination;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.apache.axiom.om.OMAbstractFactory;
import org.apache.axiom.om.OMElement;
import org.apache.axiom.om.OMNamespace;
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axiom.soap.SOAPFactory;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.AsyncConnection;
import org.apache.kudu.client.KuduClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Base64Utils;
import scala.Int;
import sun.misc.BASE64Decoder;
import sun.net.ftp.FtpClient;
import sun.net.ftp.FtpProtocolException;

import javax.annotation.Resource;
import java.io.*;
import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URL;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * <p>
 * 配置连接表 服务实现类
 * </p>
 *
 * @author zenith
 * @since 2018-08-22
 */
@Slf4j
@Service
public class ConfConnectServiceImpl extends ServiceImpl<ConfConnectMapper, ConfConnect> implements ConfConnectService {

    @Resource
    private ProcConnectMapper procConnectMapper;

    @Autowired
    private ConfConnectMapper confConnectMapper;

    @Autowired
    private AnalysisInfoMapper analysisInfoMapper;

    @Autowired
    private SysRoleDataPermissionMapper sysRoleDataPermissionMapper;

    @Autowired
    private SysTenantDataPermissionMapper sysTenantDataPermissionMapper;

    @Autowired
    private ConnectFieldInfoService connectFieldInfoService;
    @Autowired
    private SapConnectFieldService sapConnectFieldService;

    private static Object sapLockObj = new Object();
    @Autowired
    private IDepartmentService iDepartmentService;
    @Autowired
    private ISysTclDepartmentService iSysTclDepartmentService;
    @Autowired
    private TableInfoService tableInfoService;

    @Override
    public ConfConnectDetail getDetail(Long id) {
        ConfConnect confConnect = getById(id);
        if (confConnect != null) {
            ConfConnectDetail detail = new ConfConnectDetail(confConnect);
            Long userId = ShiroUtils.getUserId();
            List<DepartmentMinDto> departments = iDepartmentService.selectAllGrantDepsNoChildren(userId);
            if (StringUtils.isNotEmpty(confConnect.getDepartmentId())) {
                Set<String> idSet = departments.stream().map(DepartmentMinDto::getId).collect(Collectors.toSet());
                if (!idSet.contains(confConnect.getDepartmentId())) {
                    Integer departmentType = iDepartmentService.getDepartmentType(confConnect.getDepartmentId());
                    if(Objects.equals(departmentType, DepartmentTypeEnum.INNER.getCode())){
                        Department department = iDepartmentService.getById(confConnect.getDepartmentId());
                        if (department != null) {
                            DepartmentMinDto dto = new DepartmentMinDto();
                            BeanUtils.copyProperties(department, dto);
                            dto.setDepartmentType(DepartmentTypeEnum.INNER.getCode());
                            departments.add(dto);
                        }
                    }else{
                        SysTclDepartment sysTclDepartment = iSysTclDepartmentService.getById(confConnect.getDepartmentId());
                        if(null != sysTclDepartment){
                            DepartmentMinDto dto = new DepartmentMinDto();
                            BeanUtils.copyProperties(sysTclDepartment, dto);
                            dto.setId(sysTclDepartment.getCode());
                            dto.setDepartmentType(DepartmentTypeEnum.OUT_TCL.getCode());
                            departments.add(dto);
                        }
                    }
                }
            }
            departments = iDepartmentService.changeNewTreePack(departments);
            detail.setDepartments(departments);
            return detail;
        }
        return null;
    }

    /**
     * 判断是否包含在项目中
     *
     * @param id
     * @return
     */
    @Override
    public boolean checkIfRef(Integer id) {
        return this.procConnectMapper.selectCount(new QueryWrapper<ProcConnect>().eq("connect_id", id)) > 0;
    }

    @Override
    public boolean checkExistByName(String name, Integer id, ConfConnect confConnect) {
        //jdbc按照 数据源名字+ip+表
        LambdaQueryWrapper<ConfConnect> confConnectLambdaQueryWrapper = new LambdaQueryWrapper<>();
        confConnectLambdaQueryWrapper.eq(ConfConnect::getName,name);
        switch (confConnect.getTypeId()){
            case 1:
            case 2:
            case 3:
            case 4:
            case 10:
            case 15:
            case 17:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getHost,confConnect.getHost());
                if(StringUtils.isEmpty(confConnect.getDbname())){
                    confConnectLambdaQueryWrapper.isNull(ConfConnect ::getDbname);
                }else{
                    confConnectLambdaQueryWrapper.eq(ConfConnect ::getDbname,confConnect.getDbname());
                }
                break;
            case 5:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getHost,confConnect.getHost());
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getFtpFileLocation,confConnect.getFtpFileLocation());
                break;
            case 6:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getKafkaUrl,confConnect.getKafkaUrl());
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getTopic,confConnect.getTopic());
                break;
            case 7:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getInterfaceUrl,confConnect.getInterfaceUrl());
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getFtpFileLocation,confConnect.getFtpFileLocation());
                break;
            case 8:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getHost,confConnect.getHost());
                break;
            case 9:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getWebServiceUrl,confConnect.getWebServiceUrl());
                break;
            case 11:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getHost,confConnect.getHost());
                break;
            case 12:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getHost,confConnect.getHost());
                break;
            case 13:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getHost,confConnect.getHost());
                break;
            case 14:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getHost,confConnect.getHost());
                break;
            case 16:
                confConnectLambdaQueryWrapper.eq(ConfConnect ::getOracleNtsUrl,confConnect.getOracleNtsUrl());
                if(StringUtils.isEmpty(confConnect.getDbname())){
                    confConnectLambdaQueryWrapper.isNull(ConfConnect ::getDbname);
                }else{
                    confConnectLambdaQueryWrapper.eq(ConfConnect ::getDbname,confConnect.getDbname());
                }
                break;
        }
        if (id != null) {
            confConnectLambdaQueryWrapper.notIn(ConfConnect::getId,id);
        }
        return this.baseMapper.selectCount(confConnectLambdaQueryWrapper) > 0;

    }

    @Override
    public boolean checkPermission(Integer id) {
        ConfConnect entity = this.baseMapper.selectById(id);
        return entity.getCreatePer() == ShiroUtils.getUserId().intValue();
    }

    @Override
    public boolean checkExistById(Integer id) {
        return this.baseMapper.selectById(id) != null;
    }

    @Override
    public PageUtils selectListPage(Map<String, Object> params) {
        IPage<ConfConnect> page = this.baseMapper.selectListPage((Page<ConfConnect>) new Query<ConfConnect>().getPage(params), params);
        return new PageUtils(page);
    }


    @Override
    public PageUtils listPageByRole(Map<String, Object> params) {
        if (params.get("role_id") == null || Long.valueOf(params.get("role_id").toString()) < 1) {
            return new PageUtils<>(new ArrayList<>(),0,10,1);
        }
        DataPermissionUtil.putDataFilter(params,
                PermissionTypeEnum.RESOURCE.getType(), PermissionTypeEnum.ASSIGN.getType());
        params.put("connect_enable", 1);
//        setTenantParams(params,PermissionTypeEnum.RESOURCE.getType(),PermissionTypeEnum.ASSIGN.getType(),ShiroUtils.getTenantId());
        IPage<ConfConnect> page = this.baseMapper.listNoRelation((Page<ConfConnect>) new Query<ConfConnect>().getPage(params), params);
        Map<Integer, SysRoleDataPermission> permissionMap = Maps.newHashMap();
        Long roleId = 0L;
        if (params.get("role_id") != null) {
            roleId = Long.parseLong(params.get("role_id").toString());
            SysRoleDataPermissionExample param = new SysRoleDataPermissionExample();
            param.createCriteria()
                    .andDataTypeEqualTo(PermissionTypeEnum.RESOURCE.getType())
                    .andRoleIdEqualTo(roleId);
            List<SysRoleDataPermission> list = sysRoleDataPermissionMapper.selectByExample(param);
            if (CollectionUtils.isNotEmpty(list)) {
                for (SysRoleDataPermission sysRoleDataPermission : list) {
                    permissionMap.put(sysRoleDataPermission.getDataId(), sysRoleDataPermission);
                }
            }
        }
        List<ConfConnectPermission> dataList = new ArrayList<>();
        for (ConfConnect record : page.getRecords()) {
            if (permissionMap.get(record.getId()) != null) {
                ConfConnectPermission s = new ConfConnectPermission(record, permissionMap.get(record.getId()));
                dataList.add(s);
            } else {
                ConfConnectPermission s = new ConfConnectPermission(record, null);
                s.defaultValue(roleId, record.getId(), PermissionTypeEnum.RESOURCE.getType());
                dataList.add(s);
            }
        }
        return new PageUtils(dataList, (int) page.getTotal(), (int) page.getSize(), (int) page.getCurrent());
    }

    public List<Map<Integer,String>> userList(){
        return this.baseMapper.selectUserList(ShiroUtils.getTenantId());
    }

    public void setTenantParams(Map<String, Object> params,Integer datatype,Integer operation,Integer tenantId){
        if(tenantId!=null) {
            params.put("currentTenantId", tenantId);
            List<Integer> dataTypeList = new ArrayList<>(2);
            dataTypeList.add(datatype);
            SysTenantDataPermissionExample param = new SysTenantDataPermissionExample();
            param.createCriteria().andDataTypeIn(dataTypeList)
                    .andTenantIdEqualTo(ShiroUtils.getTenantId());
            if (operation == PermissionTypeEnum.ASSIGN.getType()) {
                param.getOredCriteria().get(0).andAssignEqualTo(1);
            }
            if (operation == PermissionTypeEnum.QUERY.getType()) {
                param.getOredCriteria().get(0).andQueryEqualTo(1);
            }
            List<SysTenantDataPermission> list = sysTenantDataPermissionMapper.selectByExample(param);
            if (CollectionUtils.isNotEmpty(list)) {
                List<Integer> dataIds = list.stream().map(i -> i.getDataId()).collect(Collectors.toList());
                params.put("dataIds", dataIds);
            }
        }
    }

    @Override
    public PageUtils listPageByTenant(Map<String, Object> params) {
        params.put("connect_enable", 1);
        params.put("currentTenantId",ShiroUtils.getTenantId());
        if(!DataPermissionUtil.isAdmin(ShiroUtils.getUserId().intValue())){
            params.put("create_per",ShiroUtils.getUserId());
        }
        IPage<ConfConnect> page = this.baseMapper.listNoRelationByTenant((Page<ConfConnect>) new Query<ConfConnect>().getPage(params), params);
        Map<Integer, SysTenantDataPermission> permissionMap = Maps.newHashMap();
        Integer tenantId = 0;
        if (params.get("tenant_id") != null) {
            tenantId = Integer.parseInt(params.get("tenant_id").toString());
            SysTenantDataPermissionExample param = new SysTenantDataPermissionExample();
            param.createCriteria()
                    .andDataTypeEqualTo(PermissionTypeEnum.RESOURCE.getType())
                    .andTenantIdEqualTo(tenantId);
            List<SysTenantDataPermission> list = sysTenantDataPermissionMapper.selectByExample(param);
            if (CollectionUtils.isNotEmpty(list)) {
                for (SysTenantDataPermission sysTenantDataPermission : list) {
                    permissionMap.put(sysTenantDataPermission.getDataId(), sysTenantDataPermission);
                }
            }
        }
        List<TenantConfConnectPermission> dataList = new ArrayList<>();
        for (ConfConnect record : page.getRecords()) {
            if (permissionMap.get(record.getId()) != null) {
                TenantConfConnectPermission s = new TenantConfConnectPermission(record, permissionMap.get(record.getId()));
                dataList.add(s);
            } else {
                TenantConfConnectPermission s = new TenantConfConnectPermission(record, null);
                s.defaultValue(tenantId, record.getId(), PermissionTypeEnum.RESOURCE.getType());
                dataList.add(s);
            }
        }
        return new PageUtils(dataList, (int) page.getTotal(), (int) page.getSize(), (int) page.getCurrent());
    }

    @Override
    public PageUtils listNoRelation(Map<String, Object> params) {
        DataPermissionUtil.putDataFilter(params,
                PermissionTypeEnum.RESOURCE.getType(), PermissionTypeEnum.QUERY.getType());
        params.put("connectEnable", 1);
//        setTenantParams(params, PermissionTypeEnum.RESOURCE.getType(),PermissionTypeEnum.QUERY.getType(),ShiroUtils.getTenantId());
        IPage<ConfConnect> page = this.baseMapper.listNoRelation((Page<ConfConnect>) new Query<ConfConnect>().getPage(params), params);
        return new PageUtils(page);
    }

    @Override
    public JSONArray getDatabaseNameToTableName(ConfConnect confConnect) {
        String host = confConnect.getHost();
        String port = confConnect.getPort();
        String userName = confConnect.getUsername();
        String passWord = confConnect.getPassword();
        String dbName = confConnect.getDbname();
        Integer typeId = confConnect.getTypeId();
        ConnectTypeEnum c = ConnectTypeEnum.ObjOf(typeId);
        JSONArray jsonArray = new JSONArray();
        //jdbc
        if (c.getType() == ConTypeType.JDBC) {
            try {
                Class.forName(c.getDriverClass());
            } catch (ClassNotFoundException e) {
                throw new RRException(DataDevelopmentBizExceptionEnum.DRIVER_LOAD_FAIL.getMessage(), DataDevelopmentBizExceptionEnum.DRIVER_LOAD_FAIL.getCode());
            }
            String url = "";
            if(StringUtils.isNotEmpty(dbName))
                url = c.getUrl().replace("<host>", host).replace("<port>", port).replace("<db_name>", dbName);
            else
                url = c.getUrl2().replace("<host>", host).replace("<port>", port);
            Map<String, Object> computer_mall = getDatabaseNameToTableName(c.getDriverClass(), url, userName, passWord);
            Set<String> tableNames = computer_mall.keySet();
            int i = 1;
            for (String t : tableNames) {
                Map<String, Object> tableNameMap = new HashMap<>();
                tableNameMap.put("id", i);
                tableNameMap.put("name", t);
                jsonArray.add(tableNameMap);
                i++;
            }
        }
        return jsonArray;
    }

    @Override
    public List<ConfConnectDto> listByCondition(ConfConnectQO confConnectQO) {
        if (null == confConnectQO || null == confConnectQO.getTypeId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_FAIL_ERROR.getMessage(), "连接的数据源类型"));
        }
        if (null == confConnectQO.getProcId()) {
            if (null == confConnectQO.getAnalysisInfoId()) {
                throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_FAIL_ERROR.getMessage(), "连接的工程id"));
            } else {
                AnalysisInfo analysisInfo = analysisInfoMapper.selectById(confConnectQO.getAnalysisInfoId());
                if (null == analysisInfo || analysisInfo.getProcId() == null) {
                    throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_FAIL_ERROR.getMessage(), "分析任务"));
                }
                confConnectQO.setProcId(analysisInfo.getProcId());
            }
        }
        // 不同的类型返回值，有可能不一样。有可以需要去再关联
        List<ConfConnect> confConnects = confConnectMapper.selectListByTypeIdAndProceId(confConnectQO);
        return PojoUtils.listConvert(ConfConnectDto.class, confConnects);
    }

    @Override
    public void checkAnalysisInfoUsed(Integer id, Integer type) {
        List<AnalysisInfo> analysisInfos = analysisInfoMapper.selectList(new QueryWrapper<AnalysisInfo>().eq("connect_id", id));
        if (type == 1) {
            //验证是否可以删除：如果有关联的分析任务，则不能删除
            if (null != analysisInfos && analysisInfos.size() > 0) {
                throw new RRException("有关联分析任务，则不能删除！");
            }
        }
        if (type == 2) {
            //验证编辑：如果有启用的分析任务，这里也不能修改。必须先禁用任务，才能进行启用
            if (null != analysisInfos && analysisInfos.size() > 0) {
                long count = analysisInfos.stream().filter(analysisInfo -> Objects.equals(true, analysisInfo.getEnable())).count();
                if (count > 0) {
                    throw new RRException("有正在运行的分析任务，则不能编辑！");
                }
            }
        }
    }

    @Override
    public List<ConfConnectDto> listAllKafka(Integer procId) {
        ConfConnectQO param = new ConfConnectQO();
        param.setTypeId(ConnectTypeEnum.Kafka.getCode());
        List<ConfConnect> confConnects;
        if (procId != null) {//项目查询  不按权限查询
            param.setProcId(procId);
            confConnects = confConnectMapper.selectListByTypeIdAndProceId(param);
        } else {//按权限查询
            Map<String, Object> params = new HashMap<>();
            params.put("page", "1");
            params.put("limit", "9999");
            params.put("typeId", ConnectTypeEnum.Kafka.getCode());
            DataPermissionUtil.putDataFilter(params,
                    PermissionTypeEnum.RESOURCE.getType(), PermissionTypeEnum.QUERY.getType());
            IPage<ConfConnect> page = this.baseMapper.listNoRelation((Page<ConfConnect>) new Query<ConfConnect>().getPage(params), params);
            confConnects = page.getRecords();
        }
        return PojoUtils.listConvert(ConfConnectDto.class, confConnects);
    }

    @Override
    public List<ConfConnect> selectQuery(Integer userId) {
        QueryWrapper<ConfConnect> confConnectQueryWrapper = new QueryWrapper<>();
        confConnectQueryWrapper.eq("connect_enable", 1);
        confConnectQueryWrapper.in("type_id", "1", "2", "3", "4", "9");
        if (!DataPermissionUtil.isAdmin(userId)) {
            List<Integer> ids = DataPermissionUtil.getDataIds(
                    userId, PermissionTypeEnum.RESOURCE.getType(), PermissionTypeEnum.QUERY.getType());
            confConnectQueryWrapper.eq("create_per", userId);
            if (CollectionUtils.isNotEmpty(ids)) {
                confConnectQueryWrapper.or();
                confConnectQueryWrapper.in("id", ids);
            }
        }
        confConnectQueryWrapper.orderByDesc("create_time");
        List<ConfConnect> confConnects = confConnectMapper.selectList(confConnectQueryWrapper);
        confConnects = CollectionUtils.isEmpty(confConnects) ? new ArrayList<>() : confConnects;
        // 增加一个默认数据源，连接到系统hive服务
        ConfConnect defaultConf = new ConfConnect();
        defaultConf.setId(0);
        defaultConf.setName("默认数据源");
        confConnects.add(0, defaultConf);
        return confConnects;
    }

    @Override
    public List<ConfConnect> selectQuery2(Integer userId) {
        List<ConfConnect> confConnects = selectQuery(userId);
        // 增加一个其他数据源，连接到系统hive服务
        ConfConnect otherConf = new ConfConnect();
        otherConf.setId(-1);
        otherConf.setName("其他数据源");
        confConnects.add(1, otherConf);
        return confConnects;
    }


    public Map getDatabaseNameToTableName(String driver, String url, String username, String password) {
        Map<String, Object> tableNameMap = new HashMap<>();

        //加载驱动
        try {
            Class.forName(driver);
        } catch (ClassNotFoundException e) {
            throw new RRException(DataDevelopmentBizExceptionEnum.DRIVER_LOAD_FAIL.getMessage(), DataDevelopmentBizExceptionEnum.DRIVER_LOAD_FAIL.getCode());
        }

        Connection connection = null;
        try {
            //获得数据库连接
            connection = DriverManager.getConnection(url, username, password);
            //获得元数据
            DatabaseMetaData metaData = connection.getMetaData();
            //获得表信息
            ResultSet tables = metaData.getTables(null, null, null, new String[]{"TABLE"});
            while (tables.next()) {
                Map<String, String> columnNameMap = new HashMap<>(); //保存字段名
                //获得表名
                String table_name = tables.getString("TABLE_NAME");
                //通过表名获得所有字段名
                ResultSet columns = metaData.getColumns(null, null, table_name, "%");
                //获得所有字段名
                while (columns.next()) {
                    //获得字段名
                    String column_name = columns.getString("COLUMN_NAME");
                    //获得字段类型
                    String type_name = columns.getString("TYPE_NAME");
                    columnNameMap.put(column_name, type_name);
                }
                tableNameMap.put(table_name, columnNameMap);
            }
        } catch (SQLException e) {
            throw new RRException(DataDevelopmentBizExceptionEnum.SQL_CONNECT_FAIL.getMessage(), DataDevelopmentBizExceptionEnum.SQL_CONNECT_FAIL.getCode());
        } finally {
            try {
                connection.close();
            } catch (SQLException e) {
                throw new RRException(DataDevelopmentBizExceptionEnum.SQL_CONNECT_CLOSE_FAIL.getMessage(), DataDevelopmentBizExceptionEnum.SQL_CONNECT_CLOSE_FAIL.getCode());
            }
        }
        return tableNameMap;
    }

    @Autowired
    private SecurityUtil securityUtil;

    /**
     * 检验是否HTTP或者HTTPS开头
     *
     * @param url
     * @return
     */
    private Boolean checkHttp(String url) {
        String[] words = new String[]{"http", "https"};
        for (String word : words) {
            if (url.contains(String.format("%s://", word))) {
                return true;
            }
        }
        return false;
    }

    private void checkWebService(ConfConnect confConnect) {
        if (StringUtils.isBlank(confConnect.getWebServiceUrl())) {
            throw new RRException("webservice的服务地址不能为空");
        }
        if (StringUtils.isBlank(confConnect.getTargetNameSpace())) {
            throw new RRException("webservice的命名空间不能为空");
        }
        if (StringUtils.isBlank(confConnect.getFunctionName())) {
            throw new RRException("webservice的方法名不能为空");
        }
    }

    private Object getObjectByAxisCallClient(ConfConnect confConnect, boolean isDebug) throws Exception {
        try {
            //验证如果是post-raw的方式，则直接ping 地址
            if(Objects.equals(confConnect.getRequestType(),3)){
                String webServiceUrl = confConnect.getWebServiceUrl();
                if (StringUtils.isEmpty(webServiceUrl)) {
                    throw new RRException("webservice的服务地址不能为空");
                }
                //截取出来对应的ip
                String host = PingIpUtil.getHost(webServiceUrl);
                if(!PingIpUtil.ping(host)){
                    throw new RRException("webservice的服务地址网络地址不对！");
                }
                return "";
            }
            //检查服务地址、命名空间、方法名是否为空
            checkWebService(confConnect);
            //生成请求的xml框架（不包括方法和参数）
            SOAPFactory fac = OMAbstractFactory.getSOAP11Factory();
            /*
             * 指定方法的命名空间，参数：
             * uri--即为wsdl文档的targetNamespace，命名空间
             * perfix--前缀
             */
            OMNamespace omNs = fac.createOMNamespace(confConnect.getTargetNameSpace(), "ure");
            // 创建方法
            OMElement method = fac.createOMElement(confConnect.getFunctionName(), omNs);
            //创建方法中的参数
            List<ConnectFieldInfo> connectFieldInfos = confConnect.getConnectFieldInfos();
            if (CollectionUtil.isNotEmpty(connectFieldInfos)) {
                for (ConnectFieldInfo connectFieldInfo : connectFieldInfos) {
                    OMElement param = fac.createOMElement(connectFieldInfo.getFieldName(), omNs);
                    param.setText(connectFieldInfo.getFieldValue());
                    //将参数放入方法中
                    method.addChild(param);
                }
            }
            //将方法放入xml中
            SOAPEnvelope env = fac.getDefaultEnvelope();
            env.getBody().addChild(method);

            //获取服务地址并处理特殊符号&，并创建http连接
            String urlString = confConnect.getWebServiceUrl();
            String newUrl = StringEscapeUtils.unescapeHtml(urlString);
            URL url2 = new URL(newUrl);
            HttpURLConnection httpConn = (HttpURLConnection) url2.openConnection();
            httpConn.setConnectTimeout(30000);//联系超时时间
            httpConn.setReadTimeout(60000);//读数据超时时间
            //将xml对象转byte
            byte[] buf = env.toString().getBytes();

            //将用户名密码处理成Authorization的值
            String user = confConnect.getUsername();
            String pwd = confConnect.getPassword();
            String auth = user + ":" + pwd;
            Base64.Encoder encoder = Base64.getEncoder();
            byte[] rel = encoder.encode(auth.getBytes());
            String res = new String(rel);


            //设置一些头参数
            httpConn.setRequestProperty("Content-Length", String.valueOf(buf.length));
            httpConn.setRequestProperty("Content-Type", "text/xml;charset=utf-8");
//        httpConn.setRequestProperty("soapActionString", confConnect.getSoapActionUrl());
            RequetTypeEnum requetTypeEnum=RequetTypeEnum.ObjOf(confConnect.getRequestType());

            if(requetTypeEnum==null){
                throw new Exception("请求类型为空");
            }
            String  requestMethod=requetTypeEnum.getName();
            if(StringUtils.isBlank(requestMethod)){
                throw new Exception("请求类型为空");
            }
            httpConn.setRequestMethod(requestMethod);
            //设置认证属性
            httpConn.setRequestProperty("Authorization", "Basic " + res);
            //输入参数和输出结果
            httpConn.setDoOutput(true);
            httpConn.setDoInput(true);

            //获取结果
            OutputStream out = httpConn.getOutputStream();
            out.write(buf);
            out.close();
            ByteArrayOutputStream outStream = new ByteArrayOutputStream();
            byte[] buffer = new byte[1024];
            int len = 0;
            while ((len = httpConn.getInputStream().read(buffer)) != -1) {
                outStream.write(buffer, 0, len);
            }
            byte[] data = outStream.toByteArray();
            outStream.close();
            httpConn.getInputStream().close();
            String result = new String(data);
//            System.out.println("result:" + result);
            return result;
        } catch (Exception e) {
            XxlJobLogger.log(e);
            throw e;
        }
    }

    //生成空的SAP数据
    private void generateNullSap(SapConnectField sapConnectField) {
        List<JSONObject> inputParams = new ArrayList<>();
        List<JSONObject> outputParams = new ArrayList<>();
        JSONObject inputParam = new JSONObject();
        JSONObject outputParam = new JSONObject();
        String[] sapParams = {"fieldName", "fieldKind", "fieldValue", "fieldAlias", "fieldInputType", "fieldDataType"};
        for (String s : sapParams) {
            inputParam.put(s, "");
            outputParam.put(s, "");
        }
        inputParams.add(inputParam);
        outputParams.add(outputParam);
        sapConnectField.setAinputParams(inputParams);
        sapConnectField.setAoutputParams(outputParams);
    }

    private ConfConnect getInfo(ConfConnect confConnect) {
        List<ConnectFieldInfo> connectFieldInfos = connectFieldInfoService.list(new QueryWrapper<ConnectFieldInfo>().eq("connect_id", confConnect.getId()));
        confConnect.setConnectFieldInfos(connectFieldInfos);

        if (confConnect.getTypeId().equals(ConnectTypeEnum.Sap.getCode())) {
            SapConnectFieldDb sapConnectFieldDb = sapConnectFieldService.getOne(new QueryWrapper<SapConnectFieldDb>().eq("connect_id", confConnect.getId()));
            if (sapConnectFieldDb != null) {
                String inputParams = sapConnectFieldDb.getAinputParams();
                String outputParams = sapConnectFieldDb.getAoutputParams();
                SapConnectField sapConnectField = new SapConnectField();
                BeanUtils.copyProperties(sapConnectFieldDb, sapConnectField);
                sapConnectField.setAinputParams(JSONObject.parseArray(inputParams, JSONObject.class));
                sapConnectField.setAoutputParams(JSONObject.parseArray(outputParams, JSONObject.class));
                confConnect.setSapConnectFieldInfo(sapConnectField);
            } else {
                SapConnectField sapConnectField = new SapConnectField();
                generateNullSap(sapConnectField);
                confConnect.setSapConnectFieldInfo(sapConnectField);
            }
        } else {
            SapConnectField sapConnectField = new SapConnectField();
            generateNullSap(sapConnectField);
            confConnect.setSapConnectFieldInfo(sapConnectField);
        }
        if (confConnect.getTypeId().equals(ConnectTypeEnum.JieKou.getCode()) || confConnect.getTypeId().equals(ConnectTypeEnum.WebService.getCode())) {
            List<ConnectFieldInfo> connectFieldInfos1 = connectFieldInfoService.list(new QueryWrapper<ConnectFieldInfo>().eq("connect_id", confConnect.getId()));
            confConnect.setConnectFieldInfos(connectFieldInfos1);
        }
        confConnect.setTypeName(ConnectTypeEnum.ObjOf(confConnect.getTypeId()).getName());
        return confConnect;
    }

    private List<Future<Boolean>> asyncConnect(ExecutorService executor, List<ConfConnect> list, Date now, boolean isDebug) throws InterruptedException {
        ExecutorService executor2 = ThreadUtil.newExecutor(1);
        CountDownLatch latch = new CountDownLatch(list.size());
        List<List<ConfConnect>> splitList = CollectionUtil.split(list, 5);
        List<Future<Boolean>> returnList = new ArrayList<>();
        for (int i = 0; i < splitList.size(); i++) {
            final int index = i;
            Future<Boolean> submit = executor.submit(() -> {
                for (ConfConnect confConnect : splitList.get(index)) {
                    try {
                        confConnect = getInfo(confConnect);

                        BaseRes res = testConnect2(executor2, confConnect, true, isDebug);
                        ConfConnect newConnect = new ConfConnect();
                        newConnect.setId(confConnect.getId());
                        newConnect.setConnectStatus(res.getCode() == 0 ? 1 : 2);//连接状态 1:正常  2:失败
                        newConnect.setModTime(now);
                        updateById(newConnect);
                    } catch (Exception e) {
                        XxlJobLogger.log(e);
                        return false;
                    } finally {
                        latch.countDown();
                    }
                }
                return true;
            });
            returnList.add(submit);
        }
        latch.await();
        executor2.shutdown();
        return returnList;
    }

    @Override
    public void testAllConnect(Integer typeId, boolean isDebug) throws InterruptedException {
        Date now = new Date();
        Map<String, Object> params = new HashMap<>();
        params.put("offset", 0);
        params.put("limit", 10000);
        params.put("sidx", "id");
        params.put("order", "desc");
        params.put("endTime", DateUtils.format(now, DateUtils.DATE_TIME_PATTERN));
        if (typeId != null)
            params.put("typeId", typeId);//测试用

//        Integer count = this.baseMapper.listAllSourceCount(params);
//        if(count == null)
//            return;

        ExecutorService executor = ThreadUtil.newExecutor(6);
        List<ConfConnect> list = this.baseMapper.listAllSource(params);
        while (list != null && list.size() > 0) {//分批循环获取
            try {
                asyncConnect(executor, list, now, isDebug);
            } catch (Exception e) {
                if (isDebug)
                    XxlJobLogger.log(e);
            }

            list = this.baseMapper.listAllSource(params);
        }
        executor.shutdown();
    }

    private BaseRes testConnect2(ExecutorService executor, ConfConnect confConnect, boolean isDecryptPwd, boolean isDebug) {
        try {
            ConnectTypeEnum t = ConnectTypeEnum.ObjOf(confConnect.getTypeId());
            //如果有密码且是密文就解密
            if (!confConnect.getTypeId().equals(ConnectTypeEnum.JieKou.getCode())) {
                if (null == confConnect.getPassword() || "".equals(confConnect.getPassword())) {
                    if (confConnect.getId() != null && !confConnect.getId().equals("")) {
                        ConfConnect confConnect1 = this.getOne(new QueryWrapper<ConfConnect>().eq("id", confConnect.getId()));
                        if (confConnect1 == null) {
                            throw new RRException("该数据源不存在");
                        }
                        if (StringUtils.isNotBlank(confConnect1.getPassword())) {
                            confConnect.setPassword(securityUtil.decrypt(confConnect1.getPassword()));
                        }

                    }
                } else {
                    if (isDecryptPwd && ( StringUtils.isNotBlank(confConnect.getPassword()))) {
;                        confConnect.setPassword(securityUtil.decrypt(confConnect.getPassword()));
                    }
                }
            }

            BaseRes res = null;
            if (t.getType() == ConTypeType.JDBC) {
                res = testJdbc(t, confConnect, isDebug);
            }else if (t.getType() == ConTypeType.OracleNTS) {
                res = testJdbcOracleNts(t, confConnect, isDebug);
            } else if (t.getType() == ConTypeType.FTP) {
                res = testFtp(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.SFTP) {
                res = testSftp(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.JieKou) {
                res = testJiekou(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.KAFKA) {
                res = testKafka(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.Sap) {
                res = testSap(executor, confConnect, isDebug);
            } else if (t.getType() == ConTypeType.WebService) {
                getObjectByAxisCallClient(confConnect, isDebug);
                res = BaseRes.ok("连接可用");
            }else if(t.getType()==ConTypeType.ES){
                res=testElasticsearch(confConnect,isDebug);
            }else if(t.getType()==ConTypeType.HBase){
                res= testHBase(confConnect,isDebug);
            }else if(t.getType()==ConTypeType.Kudu){
                res=testKudu(confConnect,isDebug);
            }else if(t.getType()==ConTypeType.MongoDB){
                res=testMongo(ConnectTypeEnum.MongoDB,confConnect,isDebug);
            }else {
                res = BaseRes.error("不支持的连接类型");
            }
            return res;
        } catch (Exception e) {
            if (e.getMessage() == null) {
                return BaseRes.error("数据源连接失败");
            }
            return BaseRes.error("数据源连接失败，" + e.getMessage());
        }
    }

    @Override
    public BaseRes testConnect(ConfConnect confConnect, boolean isDecryptPwd, boolean isDebug) {
        try {
            ConnectTypeEnum t = ConnectTypeEnum.ObjOf(confConnect.getTypeId());
            //如果有密码且是密文就解密
            if (!confConnect.getTypeId().equals(ConnectTypeEnum.JieKou.getCode())) {
                if (null == confConnect.getPassword() || "".equals(confConnect.getPassword())) {
                    if (confConnect.getId() != null && !confConnect.getId().equals("")) {
                        ConfConnect confConnect1 = this.getOne(new QueryWrapper<ConfConnect>().eq("id", confConnect.getId()));
                        if (confConnect1 == null) {
                            throw new RRException("该数据源不存在");
                        }
                        if (StringUtils.isNotBlank(confConnect1.getPassword())) {
                            //解密加密后的密码，webservice的密码也是加密的
                            confConnect.setPassword(securityUtil.decrypt(confConnect1.getPassword()));
                        }

                    }
                } else {
                    if (isDecryptPwd && (StringUtils.isNotBlank(confConnect.getPassword()))) {
                        confConnect.setPassword(securityUtil.decrypt(confConnect.getPassword()));
                    }
                }
            }
            BaseRes res = null;
            //jdbc
            if (t.getType() == ConTypeType.JDBC) {
                res = testJdbc(t, confConnect, isDebug);
            } else if (t.getType() == ConTypeType.OracleNTS) {
                res = testJdbcOracleNts(t, confConnect, isDebug);
            } else if (t.getType() == ConTypeType.FTP) {
                res = testFtp(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.SFTP) {
                res = testSftp(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.JieKou) {
                res = testJiekou(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.KAFKA) {
                res = testKafka(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.Sap) {
                res = testSap(confConnect, isDebug);
            } else if (t.getType() == ConTypeType.WebService) {
                getObjectByAxisCallClient(confConnect, isDebug);
                res = BaseRes.ok("连接可用");
            }else if (t.getType() == ConTypeType.ES) {
                //elasticsearch
                res=testElasticsearch(confConnect,isDebug);
            }else if (t.getType() == ConTypeType.HBase) {
                //HBase
                res=testHBase(confConnect,isDebug);
            } else if (t.getType() == ConTypeType.Kudu) {
                res=testKudu(confConnect,isDebug);
            } else if (t.getType() == ConTypeType.MongoDB) {
                res=testMongo(t, confConnect,isDebug);
            } else {
                res = BaseRes.error("不支持的连接类型");
            }
            return res;
        } catch (Exception e) {
            log.error("连接失败：", e);
            String cause = "";
            if (e.getMessage() == null) {
                return BaseRes.error("数据源连接失败");
            }
            StringWriter stringWriter = new StringWriter();
            e.printStackTrace(new PrintWriter(stringWriter));
            if(stringWriter.toString().contains("UnknownHostException")){
                cause = "，请求主机地址错误";
            }else if (stringWriter.toString().equals("Invalid number format for port number") || e.getMessage().contains("NumberFormatException")){
                cause = "，端口号请输入数值类型";
            }else if (e.getMessage().contains("The driver has not received any packets from the server")
                    || e.getMessage().contains("Connection refused: connect")
                    || e.getMessage().contains("Unknown host specified")
                    || e.getMessage().contains("Could not create connection to database server")){
                if (!PingIpUtil.ping(confConnect.getHost())) {
                    cause = ", 网络不通";
                } else {
                    cause = "，请确认主机地址或端口号是否填写正确";
                }
            }else if(e.getMessage().contains("Unknown user")){
                cause = "，用户名不正确";
            }else if(e.getMessage().contains("Wrong password")){
                cause = "，密码不正确";
            }else if(e.getMessage().contains("invalid username/password; logon denied") || e.getMessage().contains("authority"))
                cause="，用户名/口令无效; 登录被拒绝";
            else if(e.getMessage().contains("Invalid number format for port number"))
                cause="，连接oracle时端口号错误";
            else if(e.getMessage().contains("Network Adapter"))
                cause="，网络适配无法建立连接";
            else if(e.getMessage().contains("listener does not currently know of service"))
                cause="，监听程序当前无法识别连接描述";
            else if (e.getMessage().contains("Decryption error"))
                cause = "，请重新填写密码并保存";
            else if (confConnect.getTypeId()!=null&&!confConnect.getTypeId().equals(12)&&confConnect.getDbname() != null && e.getMessage().contains(confConnect.getDbname()))
                cause = "，数据库信息填写错误";
            else if (e.getMessage().contains("connection attempt failed"))
                cause = "，请检查端口信息";
            else if (e.getMessage().contains("timed out") || e.getMessage().contains("Timeout connecting"))
                cause = "，连接超时,请检查主机地址与端口是否正确、并确保当前主机与目标网络畅通、以及防火墙已开通";
            else if(e.getMessage().contains("webservice的服务地址网络地址不对！")){
                cause="，webservice的服务地址网络地址不对！";
            }
            else if(e.getMessage().contains("HTTP")&&e.getMessage().contains("401")){
                cause="，用户名或密码错误";
            }else if(e.getMessage().contains("Connection")&&e.getMessage().contains("refused")){
                cause="，请检查主机地址与端口";
            }else if(e.getMessage().contains("unknown") && e.getMessage().contains("error")) {
                cause="，请检查主机地址与端口";
            }else if ((confConnect.getUsername() != null && e.getMessage().contains(confConnect.getUsername())) || (confConnect.getPassword() != null && e.getMessage().contains(confConnect.getPassword()))){
                cause = "，用户名或密码错误";
            }else if(e.getMessage().contains("make sure URL is format for http")){
                cause = "，请检查请求主机地址";
            }else if(e.getMessage().contains("port out of range")){
                cause="，请端口是否正确";
            }else if(e.getMessage().contains("port")){
                cause="，请检查主机地址与端口";
            }else if(e.getMessage().contains("Unable to parse response body")){
                cause="，请检查主机地址";
            }else if(e.getMessage().contains("Connection refused")){
                cause="，请检查端口";
            }else{
                cause="，"+e.getMessage();
            }
            return BaseRes.error("数据源连接失败" + cause);
        }
    }

    private BaseRes testJdbcOracleNts(ConnectTypeEnum t, ConfConnect confConnect, boolean isDebug) throws Exception {
        //暂时直接验证通过。待有环境再测试
        return BaseRes.ok("连接成功");
//        Connection conn = null;
//        try {
//            Class.forName(t.getDriverClass());
//            DriverManager.setLoginTimeout(2);
//            String URL = confConnect.getOracleNtsUrl();
//            conn = DriverManager.getConnection(URL, confConnect.getUsername(), confConnect.getPassword());
//            if (conn != null && !conn.isClosed()){
//                return BaseRes.ok("连接成功");
//            }
//        } catch (Exception e) {
//            if (isDebug)
//                XxlJobLogger.log(e);
//            throw e;
//        } finally {
//            if (conn != null)
//                conn.close();
//        }
//        return BaseRes.error("连接不可用");
    }

    private BaseRes testJdbc(ConnectTypeEnum t, ConfConnect confConnect, boolean isDebug) throws Exception {
        Connection conn = null;
        try {
            Class.forName(t.getDriverClass());
            DriverManager.setLoginTimeout(2);
            String URL = "";
            if(StringUtils.isEmpty(confConnect.getDbname())){
                URL = t.getUrl2().replace("<host>", confConnect.getHost()).replace("<port>", confConnect.getPort());
            }else {


                URL = t.getUrl().replace("<host>", confConnect.getHost()).replace("<port>", confConnect.getPort()).replace("<db_name>", confConnect.getDbname());
            }

            conn = DriverManager.getConnection(URL, confConnect.getUsername(), confConnect.getPassword());
            if (conn != null && !conn.isClosed()){
                return BaseRes.ok("连接成功");
            }
        } catch (Exception e) {
            if (isDebug)
                XxlJobLogger.log(e);
            throw e;
        } finally {
            if (conn != null)
                conn.close();
        }
        return BaseRes.error("连接不可用");
    }

    private BaseRes testFtp(ConfConnect confConnect, boolean isDebug) throws Exception {
        //ftp
        FtpClient ftpClient = null;
        try {
            SocketAddress addr = new InetSocketAddress(confConnect.getHost(), Integer.parseInt(confConnect.getPort()));
            ftpClient = FtpClient.create();
            ftpClient.setConnectTimeout(5*1000);
            ftpClient.connect(addr);
            ftpClient.login(confConnect.getUsername(), confConnect.getPassword().toCharArray());
            try {
                ftpClient.changeDirectory(confConnect.getFtpFileLocation());
            } catch (FtpProtocolException | IOException e) {
                if (e.getMessage().contains("Access is denied")) {
                    return BaseRes.error(confConnect.getFtpFileLocation() + "目录无权访问");
                }
                log.error("文件路径有误:", e);
                return BaseRes.error(confConnect.getFtpFileLocation() + "文件路径有误");
            }
        } catch (Exception e) {
            if (isDebug)
                XxlJobLogger.log(e);
            throw e;
        } finally {
            if (ftpClient != null)
                ftpClient.close();
        }
        return BaseRes.ok();
    }

    private BaseRes testSftp(ConfConnect confConnect, boolean isDebug)throws Exception {
        //sftp
        Session session = null;
        ChannelSftp sftp = null;
        try {

            JSch jSch = new JSch();
            session = jSch.getSession(confConnect.getUsername(), confConnect.getHost(),  Integer.parseInt(confConnect.getPort()));
            session.setTimeout(5*1000);

            if (!StringUtils.isEmpty(confConnect.getPassword())) {
                session.setPassword(confConnect.getPassword());
            }
            Properties config = new Properties();
            config.put("StrictHostKeyChecking", "no");

            session.setConfig(config);
            session.connect();

            Channel channel = session.openChannel("sftp");
            channel.connect();

            sftp = (ChannelSftp) channel;

            sftp.cd(confConnect.getFtpFileLocation());

        } catch (JSchException e) {
            if (isDebug)
                XxlJobLogger.log(e);
            return BaseRes.error("认证失败");
        } catch (SftpException e) {
            return BaseRes.error(confConnect.getFtpFileLocation() + "文件路径有误");
        } finally {
            if (sftp != null)
                sftp.disconnect();
        }
        return BaseRes.ok();
    }

    private BaseRes testJiekou(ConfConnect confConnect2, boolean isDebug) throws Exception {
        Integer result = 500;
        String restMsg = "";
        ConfConnect confConnect1 =testJiekouParamsUpdate(confConnect2);
        String format = confConnect1.getFormat();//请求格式text或者json
        String url = confConnect1.getTmeInterfaceUrl();//这里数据源接口的时候也包含端口
        if(StringUtils.isEmpty(url)){
            url = confConnect1.getInterfaceUrl();
        }
        int requestType = confConnect1.getRequestType();//get或者post
        RequetTypeEnum requetTypeEnum = cn.getech.data.development.constant.RequetTypeEnum.ObjOf(requestType);
        int code = confConnect1.getCode();
        if (!checkHttp(url)) {
            return BaseRes.error("您输入的URL不合规");
        }
        Method method = Method.valueOf(requetTypeEnum.getName());
        List<ConnectFieldInfo> connectFieldInfos = confConnect1.getConnectFieldInfos();
        try {
            HttpRequest request = HttpUtil.createRequest(method, url).contentType(String.format("application/%s;charset=%s", format, CodeType.ObjOf(code).getName()));
            // body he header
            JSONObject bodyJson = new JSONObject();
            Map<String, String> headerJson = new HashMap<>();
            if (connectFieldInfos != null) {
                connectFieldInfos.forEach(c -> {
                    switch (c.getFieldType()) {
                        // 0:body 里面
                        case 0:
                            bodyJson.put(c.getFieldName(), c.getFieldValue());
                            break;
                        // 1：header里面
                        case 1:
                            headerJson.put(c.getFieldName(), c.getFieldValue());
                            break;
                        default:
                            break;
                    }
                });
            }
            request.addHeaders(headerJson);
            if (bodyJson.size() > 0) {
                request.body(bodyJson.toJSONString());
            }
            request.setConnectionTimeout(60000); // 设置连接主机超时
            request.setReadTimeout(60000);
            HttpResponse execute = request.execute();
            if (null == execute) {
                return BaseRes.error("连接不可用！");
            }
            result = execute.getStatus();
            restMsg = execute.body();
        } catch (Exception ex) {
            if (isDebug)
                XxlJobLogger.log(ex);
            throw ex;
        }
        if (200 != result) {
            if (405 == result) {
                return BaseRes.error("参数错误！");
            }
            if (404 == result) {
                return BaseRes.error("接口路径不存在！");
            }
            return BaseRes.error("连接不可用！");
        }else{
            if(StringUtils.isNotEmpty(restMsg)){
                JSONObject jsonObject = JSONObject.parseObject(restMsg);
                if(null != jsonObject){
                    Integer code1 = jsonObject.getInteger("code");
                    if(null != code1 && 200 != code1 && code1 != 0 && code1 != 20000){
                        if(code1 == 401){
                            return BaseRes.error(500,jsonObject.getString("msg"));
                        }
                        if(jsonObject.getString("msg").contains("未知异常")){
                            return BaseRes.error(code1,"连接失败请检查该服务是否可用,参数是否正确");
                        }
                        return BaseRes.error(code1,jsonObject.getString("msg"));
                    }
                }
            }
        }
        return BaseRes.okWithData(result);
    }

    private BaseRes testKafka(ConfConnect confConnect, boolean isDebug) throws Exception {
        Set data = LocalKafkaUtil.topicList(confConnect.getKafkaUrl());
        if (CollectionUtils.isNotEmpty(data)) {
            return BaseRes.ok("连接可用");
        } else {
            return BaseRes.error("连接不可用");
        }
    }

    /**
     * 测试es 是否可用
     */
    private BaseRes testElasticsearch(ConfConnect confConnect, boolean isDebug) throws Exception {
        try {
            EsDto esDto=new EsDto();
            BeanUtil.copyProperties(confConnect,esDto);
            esDto.setClusterName(confConnect.getClusterName());
            esDto.setIp(confConnect.getHost());
            esDto.setPort(confConnect.getPort());
            esDto.setUsername(confConnect.getUsername());
            esDto.setPassword(confConnect.getPassword());
            esDto.setDbname(confConnect.getDbname());
            try{
                RestHighLevelClient transportClient= EsUtil.getHighClient(esDto);
                if(transportClient!=null){
                    try{
                        transportClient.close();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    return BaseRes.ok("连接可用");
                }else{
                     throw new Exception("请检查主机与端口");
                }
            }catch (Exception e){
                 throw  e;

            }



        } catch (Exception e) {
                XxlJobLogger.log(e);
            throw e;
        }
    }

    /**
     * 测试es 是否可用
     */
    private BaseRes testHBase(ConfConnect confConnect, boolean isDebug) throws Exception {
        try {


            HBaseDto hBaseDto=new HBaseDto();
            BeanUtil.copyProperties(confConnect,hBaseDto);
            hBaseDto.setIp(confConnect.getHost());
            hBaseDto.setPort(confConnect.getPort());
            hBaseDto.setUsername(confConnect.getUsername());
            hBaseDto.setPassword(confConnect.getPassword());
            AsyncConnection connection= HBaseUtil.getConnection(hBaseDto);
            if(connection!=null){
                try{
                    connection.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
                return BaseRes.ok("连接可用");
            }else{
                throw new Exception("请检查主机与端口");
            }

        } catch (Exception e) {
            if (isDebug)
                XxlJobLogger.log(e);
            throw e;
        }
    }

    /**
     * 测试es 是否可用
     */
    private BaseRes testKudu(ConfConnect confConnect, boolean isDebug) throws Exception {
        try {
            KuduDto kuduDto=new KuduDto();
            BeanUtil.copyProperties(confConnect,kuduDto);
            kuduDto.setIp(confConnect.getHost());
            kuduDto.setPort(confConnect.getPort());
            kuduDto.setUsername(confConnect.getUsername());
            kuduDto.setPassword(confConnect.getPassword());
            KuduClient kuduClient= KuduUtil.getKuduClient(kuduDto);
            if(kuduClient!=null){
                try{
                    kuduClient.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
                return BaseRes.ok("连接可用");
            }else{
                throw new Exception("请检查主机与端口");
            }

        } catch (Exception e) {
            if (isDebug)
                XxlJobLogger.log(e);
            throw e;
        }
    }

    private BaseRes testSap(ConfConnect confConnect, boolean isDebug) throws Exception {
        try {
            SapConnectField sapConnectFieldInfo = confConnect.getSapConnectFieldInfo();
            if (sapConnectFieldInfo.getIsGroup() == null || sapConnectFieldInfo.getIsGroup() == 0) {
                if (!("".equals(sapConnectFieldInfo.getR3Name()) || "".equals(sapConnectFieldInfo.getGroupName()) || sapConnectFieldInfo.getR3Name() == null || sapConnectFieldInfo.getGroupName() == null)) {
                    return BaseRes.error("不是组查询时组名称或者R3Name不为空");
                }
            } else if (sapConnectFieldInfo.getIsGroup() != null && sapConnectFieldInfo.getIsGroup() == 1) {
                if (sapConnectFieldInfo.getR3Name() == null || "".equals(sapConnectFieldInfo.getR3Name()) || sapConnectFieldInfo.getGroupName() == null || "".equals(sapConnectFieldInfo.getGroupName())) {
                    return BaseRes.error("组查询时组名称或者R3Name为空");
                }
            } else {
                return BaseRes.error("是否为集群参数格式错误");
            }
            JCoDestination dest = SAPConnUtils.connect(confConnect);
//            dest.ping();
            return BaseRes.ok("连接可用");
        } catch (Exception e) {
            if (isDebug)
                XxlJobLogger.log(e);
            throw e;
        }
    }

    private BaseRes testSap(ExecutorService executor, ConfConnect confConnect, boolean isDebug) throws Exception {
        try {
            SapConnectField sapConnectFieldInfo = confConnect.getSapConnectFieldInfo();
            if (sapConnectFieldInfo.getIsGroup() == null || sapConnectFieldInfo.getIsGroup() == 0) {
                if (!("".equals(sapConnectFieldInfo.getR3Name()) || "".equals(sapConnectFieldInfo.getGroupName()) || sapConnectFieldInfo.getR3Name() == null || sapConnectFieldInfo.getGroupName() == null)) {
                    return BaseRes.error("不是组查询时组名称或者R3Name不为空");
                }
            } else if (sapConnectFieldInfo.getIsGroup() != null && sapConnectFieldInfo.getIsGroup() == 1) {
                if (sapConnectFieldInfo.getR3Name() == null || "".equals(sapConnectFieldInfo.getR3Name()) || sapConnectFieldInfo.getGroupName() == null || "".equals(sapConnectFieldInfo.getGroupName())) {
                    return BaseRes.error("组查询时组名称或者R3Name为空");
                }
            } else {
                return BaseRes.error("是否为集群参数格式错误");
            }
            boolean returnSubmit = false;
            synchronized (sapLockObj) {
                Future<Boolean> submit = executor.submit(() -> {
                    try {
                        JCoDestination dest = SAPConnUtils.connect(confConnect);
                    } catch (Exception e) {
                        return false;
                    }
                    return true;
                });
                returnSubmit = submit.get(20, TimeUnit.SECONDS);
            }
//            dest.ping();
            if (returnSubmit)
                return BaseRes.ok("连接可用");
            return BaseRes.error("连接不可用");
        } catch (Exception e) {
            if (isDebug)
                XxlJobLogger.log(e);
            throw e;
        }
    }

    /**
     * 测试Mongo 是否可用
     */
    private BaseRes testMongo(ConnectTypeEnum t,ConfConnect confConnect, boolean isDebug) throws Exception {
        MongoClient mongoClient = null;
        try {
            String uriStr = "mongodb://" + confConnect.getUsername() + ":" + confConnect.getPassword() + "@" +
                    confConnect.getHost() + ":" + confConnect.getPort() ;
            MongoClientURI uri = new MongoClientURI(uriStr);
            mongoClient = new MongoClient(uri);
            List<String> list = mongoClient.getDatabaseNames();
            if(list!=null){
                return BaseRes.ok("连接成功");
            }
        } catch (Exception e) {
            if (isDebug)
                XxlJobLogger.log(e);
            throw e;
        } finally {
            if (mongoClient != null)
                mongoClient.close();
        }
        return BaseRes.error("连接不可用");
    }

    @Override
    public BaseRes testConnect(ConfConnect confConnect, boolean isDecryptPwd) {
        return testConnect(confConnect, isDecryptPwd, false);
    }

    @Override
    public PageUtils queryPage(Map<String, Object> params) {
        DataPermissionUtil.putDataFilter(params,
                PermissionTypeEnum.RESOURCE.getType(), PermissionTypeEnum.QUERY.getType());
        if(params.get("databaseType")!=null){
            switch ((String) params.get("databaseType")){
                case "MySQL":params.put("typeId",1);break;
                case "Oracle":params.put("typeId",2);break;
                case "SQL_Server":params.put("typeId",3);break;
                case "PostgreSQL":params.put("typeId",4);break;
                case "FTP":params.put("typeId",5);break;
                case "Kafka":params.put("typeId",6);break;
                case "JieKou":params.put("typeId",7);break;
                case "Sap":params.put("typeId",8);break;
                case "WebService":params.put("typeId",9);break;
                case "SAPHANA":params.put("typeId",10);break;
                case "Hive":params.put("typeId",11);break;
                case "ElasticSearch":params.put("typeId",12);break;
                case "HBase":params.put("typeId",13);break;
                case "Kudu":params.put("typeId",14);break;
                case "ClickHouse":params.put("typeId",15);break;
                case "MongoDB":params.put("typeId",17);break;
                case "SFTP":params.put("typeId",18);break;
            }
        }
        params.put("currentTenantId",ShiroUtils.getTenantId());
//        setTenantParams(params,PermissionTypeEnum.RESOURCE.getType(),PermissionTypeEnum.QUERY.getType(),ShiroUtils.getTenantId());
        IPage<ConfConnect> page = this.baseMapper.listNoRelation((Page<ConfConnect>) new Query<ConfConnect>().getPage(params), params);
//        String key = (String) params.get("name");
//        Integer create_per = (Integer) params.get("create_per");
//        String type_id = (String) params.get("typeId");
//        IPage<ConfConnect> page = this.page(
//                new Query<ConfConnect>().getPage(params),
//                new QueryWrapper<ConfConnect>().like(StringUtils.isNotBlank(key), "name", key)
//                        .eq(StringUtils.isNotBlank(type_id), "type_id", type_id)
//                        .eq("create_per", create_per)
//
//        );
        return new PageUtils(page);
    }

    @Override
    public ConfConnect selectOne(Integer id) {
        ConfConnect confConnect;
        QueryWrapper<ConfConnect> queryWrapper = new QueryWrapper();
        queryWrapper.eq("id", id);
        confConnect = this.baseMapper.selectOne(queryWrapper);
        if (confConnect != null) {
            return confConnect;
        }
        return confConnect;
    }

    @Override
    public List<ConfConnect> selectByNameAndProId(Map<String, Object> params) {
        List<ConfConnect> confConnects;
        confConnects = this.confConnectMapper.selectByNameAndProId(params);
        if (confConnects != null && confConnects.size() > 0) {
            /**添加字段值*/
            confConnects.forEach(x -> {
                List<ConnectFieldInfo> connectFieldInfos = connectFieldInfoService.list(new QueryWrapper<ConnectFieldInfo>().eq("connect_id", x.getId()));
                x.setConnectFieldInfos(connectFieldInfos);
            });
        }
        return confConnects;
    }

    @Override
    public boolean checkAdminPermission(Integer id, Integer operation) {
        Integer tenantId = ShiroUtils.getTenantId();
        //如果是租户管理员，则判断这个数据源的主租户和当前租户是否相同
        ConfConnect confConnect = confConnectMapper.selectById(id);
        if(null == confConnect){
            return false;
        }
        if(Objects.equals(tenantId,confConnect.getTenantId())){
            return true;
        }
        //通过租户授权的权限
        SysTenantDataPermissionExample param = new SysTenantDataPermissionExample();
        SysTenantDataPermissionExample.Criteria criteria = param.createCriteria();
        criteria.andTenantIdEqualTo(tenantId).andDataTypeEqualTo(PermissionTypeEnum.RESOURCE.getType());
        if (PermissionTypeEnum.QUERY.getType().equals(operation)) {
            criteria.andQueryEqualTo(YesOrNoEnum.YES.getValue());
        } else if (PermissionTypeEnum.UPDATE.getType().equals(operation)) {
            criteria.andUpdateEqualTo(YesOrNoEnum.YES.getValue());
        } else if (PermissionTypeEnum.DELETE.getType().equals(operation)) {
            criteria.andDeleteEqualTo(YesOrNoEnum.YES.getValue());
        } else if (PermissionTypeEnum.CREATE.getType().equals(operation)) {
            criteria.andCreateEqualTo(YesOrNoEnum.YES.getValue());
        } else if (PermissionTypeEnum.ASSIGN.getType().equals(operation)) {
            criteria.andAssignEqualTo(YesOrNoEnum.YES.getValue());
        } else if (PermissionTypeEnum.EXPORT.getType().equals(operation)) {
            criteria.andExportEqualTo(YesOrNoEnum.YES.getValue());
        } else if (PermissionTypeEnum.DECRYPT.getType().equals(operation)) {
            criteria.andDecryptEqualTo(YesOrNoEnum.YES.getValue());
        }else if (PermissionTypeEnum.WRITE.getType().equals(operation)) {
            criteria.andWriteEqualTo(YesOrNoEnum.YES.getValue());
        }
        List<SysTenantDataPermission> list = sysTenantDataPermissionMapper.selectByExample(param);
        if(CollectionUtil.isNotEmpty(list)){
            List<Integer> collect = list.stream().map(SysTenantDataPermission::getDataId).collect(Collectors.toList());
            if(collect.contains(id)){
                return true;
            }
        }
        return false;
    }

    @Override
    public List<Integer> queryIdsByProjectId(Integer projectId) {
        return this.baseMapper.queryIdsByProjectId( projectId);
    }

    @Override
    public Integer selectByName(String name) {
        return this.baseMapper.selectByName(name);
    }


    @Override
    public List<String> getMetaDbNames(Integer id) {
        ConfConnect confConnect = getById(id);
        String sourceConf = null;
        if (null != confConnect && confConnect.getHost() != null && confConnect.getPort() != null) {
            ConnectTypeEnum t = ConnectTypeEnum.ObjOf(confConnect.getTypeId());
            if (t != null) {
                sourceConf = t.getUrl().replace("<host>", confConnect.getHost())
                        .replace("<port>", confConnect.getPort())
                        .replace("<db_name>", "");
            }
        }
        Integer tenantId = ShiroUtils.getTenantId();
        return confConnectMapper.getMetaDbNames(id,sourceConf,tenantId);
    }

    @Override
    public List<String> getDbnames(ConfConnect confConnect) {
        ConnectTypeEnum t = ConnectTypeEnum.ObjOf(confConnect.getTypeId());

        List<String> dbnames = new ArrayList<>();
        if (StringUtils.isNotEmpty(confConnect.getDbname())) {
            dbnames.add(confConnect.getDbname());
        } else {
            if (t.getType() != ConTypeType.JDBC) {
                return null;
            }

            Connection conn = null;
            try {
                Class.forName(t.getDriverClass());
                String URL = t.getUrl2().replace("<host>", confConnect.getHost()).replace("<port>", confConnect.getPort());
                String password = securityUtil.decrypt(confConnect.getPassword());
                conn = DriverManager.getConnection(URL, confConnect.getUsername(), password);
                if(confConnect.getTypeId().equals(ConnectTypeEnum.ClickHouse.getCode())||
                confConnect.getTypeId().equals(ConnectTypeEnum.Hive.getCode())
                ){
                    Statement statement = conn.createStatement();
                    ResultSet results = statement.executeQuery("show databases");
                    ResultSetMetaData rsmd = results.getMetaData();
                    while (results.next()) {
                        for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                            dbnames.add(results.getString(rsmd.getColumnName(i)));
                        }
                    }
                }else {
                    ResultSet resultSet = conn.getMetaData().getCatalogs();
                    while (resultSet.next()){
                        String table_cat = resultSet.getString("TABLE_CAT");
                        dbnames.add(table_cat);
                    }
                }
            } catch (Exception e) {
                log.error("", e);
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (SQLException e) {
                        log.error("", e);
                    }
                }
            }
        }

        return dbnames;
    }
    @Override
    public List<TableInfoFieldRes> getTableInfoField(TableInfoFieldReq req) {
        Integer tenantId = ShiroUtils.getTenantId();
        req.setTenantId(tenantId);
        if (req.getSourceId() == null) {
            req.setSourceId(0);
        }
        //先查询表，再根据表查询对应的表结构
        List<TableInfoFieldRes>  sourceTables = new ArrayList<>();
        Integer sourceId = req.getSourceId();
        if(sourceId <= 0){
            if (DefConnectEnum.HIVE.getCode().equals(sourceId)) {
                req.setSourceId(0);
                req.setTableType(1);//离线表
                req.setStoreEngine(1);
            }
            if (DefConnectEnum.IMPALA.getCode().equals(sourceId)) {
                req.setSourceId(0);
                req.setTableType(1);//离线表
            }
            if (DefConnectEnum.KAFKA.getCode().equals(sourceId)) {
                req.setSourceId(0);
                req.setTableType(2);//实时表
            }
            sourceTables = confConnectMapper.getMetaTables(req);

        }else{
            //拼装成ipconif 进行查询
            ConfConnect confConnect = getById(req.getSourceId());
            if(null != confConnect){
                String dbName = "";
                if(null != req.getDbName()){
                    dbName = req.getDbName();
                }
                ConnectTypeEnum t = ConnectTypeEnum.ObjOf(confConnect.getTypeId());
                String sourceConf = t.getUrl().replace("<host>", confConnect.getHost()).replace("<port>", confConnect.getPort()).replace("<db_name>", dbName);
                sourceTables =  confConnectMapper.getMetaTablesByConfig(sourceConf,req.getSourceId(),tenantId);
            }

        }
        if (CollectionUtil.isEmpty(sourceTables)) {
            return sourceTables;
        }
        List<TableRelationshipEntity> shipList = new ArrayList<>();
        //根据元数据表，获取对应的表结构
        List<Integer> tableIds = sourceTables.stream().map(TableInfoFieldRes::getId).distinct().collect(Collectors.toList());
        List<TableRelationshipEntity> fields = confConnectMapper.getMetaTableFields(tableIds,req.getCurrentTableId());
        ConfConnect confConnect = confConnectMapper.selectById(req.getSourceId());
        for (TableRelationshipEntity field : fields) {
            TableRelationshipEntity relationshipEntity = new TableRelationshipEntity();
            relationshipEntity.setRelationFieldName(field.getRelationFieldName());
            relationshipEntity.setRelationFieldType(field.getRelationFieldType());
            relationshipEntity.setRelationFieldDesc(field.getRelationFieldDesc());
            relationshipEntity.setRelationTableName(field.getRelationTableName());
            relationshipEntity.setRelationTableDesc(null);
            relationshipEntity.setRelationDbname(req.getDbName());
            relationshipEntity.setRelationCollTableId(field.getRelationCollTableId());
            relationshipEntity.setRelationSourceId(req.getSourceId());
            if (confConnect != null) {
                relationshipEntity.setRelationSourceName(confConnect.getName());
            }
            shipList.add(relationshipEntity);
        }
        Map<Integer, List<TableRelationshipEntity>> shipListMap
                = shipList.stream().filter(s -> s.getRelationCollTableId()!=null)
                .collect(Collectors.groupingBy(TableRelationshipEntity::getRelationCollTableId));
        List<TableInfoFieldRes> restList = new ArrayList<>();
        for (TableInfoFieldRes tableInfo : sourceTables) {
            List<TableRelationshipEntity> shipListTable = shipListMap.get(tableInfo.getId());
            if (CollectionUtils.isNotEmpty(shipListTable)) {
                for (TableRelationshipEntity tableRelationshipEntity : shipListTable) {
                    tableRelationshipEntity.setRelationTableDesc(tableInfo.getDesc());
                }
                tableInfo.setRelationshipes(shipListTable);
            } else {
                tableInfo.setRelationshipes(new ArrayList<>());
            }
            restList.add(tableInfo);
        }
        return restList;
    }
    @Override
    public TableInfoFieldRelationRes getTableInfoField(TableInfoFieldRelationReq req) {
        //默认数据源的表
        TableInfoFieldRelationRes tableInfo = new TableInfoFieldRelationRes();
        String sourceName = null;
        ConfConnect confConnect = null;
        TableInfo table = confConnectMapper.selectTableByName(req);
        if (table != null) {
            List<TableField> fieldList = confConnectMapper.selectListByTableId(table.getId());
            tableInfo.setTableName(table.getTableName());
            tableInfo.setDesc(table.getDesc());
            tableInfo.setTableAlias(table.getTableAlias());
            confConnect = confConnectMapper.selectById(table.getSourceId());
            Integer sourceId = table.getSourceId();
            for (TableField tableField : fieldList) {
                TableRelationshipEntity relationshipEntity = new TableRelationshipEntity();
                List<TableRelationshipEntity> relationshipEntities = new ArrayList<>();
                String typeName = tableField.getFieldType();
                if (confConnect != null) {
                    typeName = DataTypeTransfer.transferDataType(typeName.toUpperCase(), confConnect.getTypeId());
                    relationshipEntity.setRelationSourceName(confConnect.getName());
                } else {

                    if (Integer.valueOf(0).equals(sourceId) &&
                            Integer.valueOf(1).equals(table.getTableType()) &&
                            Integer.valueOf(1).equals(table.getStoreEngine())
                    ) {
                        sourceName = DefConnectEnum.HIVE.getName();
                    } else if (Integer.valueOf(0).equals(sourceId)
                            && Integer.valueOf(1).equals(table.getTableType())) {
                        sourceId = DefConnectEnum.IMPALA.getCode();
                        sourceName = DefConnectEnum.IMPALA.getName();
                    } else if (Integer.valueOf(0).equals(sourceId)
                            && Integer.valueOf(2).equals(table.getTableType())) {
                        sourceName = DefConnectEnum.KAFKA.getName();
                        sourceId = DefConnectEnum.KAFKA.getCode();
                    } else if (DefConnectEnum.OTHER.getCode().equals(sourceId)
                            && Integer.valueOf(3).equals(table.getTableType())) {
                        sourceName = DefConnectEnum.OTHER.getName();
                    }
                    relationshipEntity.setRelationSourceName(sourceName);
                }
                if (StringUtils.isEmpty(tableField.getFieldAlias())) {
                    tableField.setFieldAlias(tableField.getFieldName());
                    relationshipEntity.setRelationFieldDesc(tableField.getFieldName());
                } else {
                    relationshipEntity.setRelationFieldDesc(tableField.getFieldAlias());
                }
                tableField.setFieldType(typeName);
                relationshipEntity.setRelationFieldName(tableField.getFieldName());
                relationshipEntity.setRelationFieldType(tableField.getFieldType());
                relationshipEntity.setRelationTableName(table.getTableName());
                relationshipEntity.setRelationTableDesc(table.getDesc());
                relationshipEntity.setRelationDbname(table.getDbName());
                relationshipEntity.setRelationSourceId(sourceId);
                relationshipEntity.setRelationCollTableId(table.getId());
                relationshipEntities.add(relationshipEntity);
                tableField.setRelationshipes(relationshipEntities);
            }
            tableInfo.setTableFieldInfos(fieldList);
            return tableInfo;
        }
        return tableInfo;
    }

    @Override
    public PageUtils pageByNameAndProId(Map<String, Object> params) {
//        DataPermissionUtil.putDataFilter(params,sysRoleDataPermissionMapper,
//                PermissionTypeEnum.RESOURCE.getType(),PermissionTypeEnum.QUERY.getType());
        IPage<ConfConnect> page = this.baseMapper.pageByNameAndProId((Page<ConfConnect>) new Query<ConfConnect>().getPage(params), params);
        return new PageUtils(page);
    }

    private Connection getConnection(ConfConnect confConnect) throws Exception{
        ConnectTypeEnum t = ConnectTypeEnum.ObjOf(confConnect.getTypeId());
        Class.forName(t.getDriverClass());
        DriverManager.setLoginTimeout(2);
        String URL = null;
        if(StringUtils.isEmpty(confConnect.getDbname())){
            URL = t.getUrl2().replace("<host>", confConnect.getHost()).replace("<port>", confConnect.getPort());
        }else {
            URL = t.getUrl().replace("<host>", confConnect.getHost()).replace("<port>", confConnect.getPort()).replace("<db_name>", confConnect.getDbname());
        }
        Connection conn = DriverManager.getConnection(URL, confConnect.getUsername(), confConnect.getPassword());
        return conn;
    }

    private List<TableInfoFieldRes> getSourceTables(ConfConnect confConnect, TableInfoFieldReq req) {
        List<TableInfoFieldRes> list = new ArrayList<>();

        Connection conn = null;
        try {
            conn = getConnection(confConnect);

            String[] types = new String[1];
            types[0] = "TABLE";
            if(StringUtils.isEmpty(confConnect.getDbname()))
                conn.setCatalog(req.getDbName());
            ResultSet result = conn.getMetaData().getTables(null, null, null, types);
            while (result.next()) {
                if (result != null) {
                    TableInfoFieldRes tableInfo = new TableInfoFieldRes();
                    tableInfo.setTableName(result.getString("TABLE_NAME"));
                    String remarks = result.getString("REMARKS");
                    tableInfo.setDesc(remarks);
                    if(StringUtils.isEmpty(remarks))
                        tableInfo.setTableAlias(tableInfo.getTableName());
                    else
                        tableInfo.setTableAlias(remarks);
                    list.add(tableInfo);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        } finally {
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException e) {
                log.error(e.getMessage());
            }
        }

        return list;
    }

    private List<TableInfoFieldRelationRes> getSourceTables(ConfConnect confConnect, TableInfoFieldRelationReq req) {
        List<TableInfoFieldRelationRes> list = new ArrayList<>();

        Connection conn = null;
        try {
            conn = getConnection(confConnect);

            String[] types = new String[1];
            types[0] = "TABLE";
            if(StringUtils.isEmpty(confConnect.getDbname()))
                conn.setCatalog(req.getDbName());
            String tableNamePattern = null;
            if(StringUtils.isNotEmpty(req.getTableName()))
                tableNamePattern = req.getTableName();
            ResultSet result = conn.getMetaData().getTables(null, null, tableNamePattern, types);
            while (result.next()) {
                if (result != null) {
                    TableInfoFieldRelationRes tableInfo = new TableInfoFieldRelationRes();
                    tableInfo.setTableName(result.getString("TABLE_NAME"));
                    String remarks = result.getString("REMARKS");
                    tableInfo.setDesc(remarks);
                    if(StringUtils.isEmpty(remarks))
                        tableInfo.setTableAlias(tableInfo.getTableName());
                    else
                        tableInfo.setTableAlias(remarks);
                    list.add(tableInfo);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        } finally {
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException e) {
                log.error(e.getMessage());
            }
        }

        return list;
    }

    private  List<TableRelationshipEntity> getSourceColumns(ConfConnect confConnect, TableInfoFieldReq req) {
        List<TableRelationshipEntity> shipList = new ArrayList<>();
        Connection conn = null;
        try {
            conn = getConnection(confConnect);
            if (conn == null)
                throw new RRException("获取数据库连接失败");

            if(StringUtils.isEmpty(confConnect.getDbname()))
                conn.setCatalog(req.getDbName());


            ResultSet result = conn.getMetaData().getColumns(null, null, null, null);
            while (result.next()) {
                if (result != null) {
                    String columnName = result.getString("COLUMN_NAME");
                    String typeName = result.getString("TYPE_NAME");
                    String remarks = result.getString("REMARKS");
                    String tableName = result.getString("TABLE_NAME");
                    TableRelationshipEntity relationshipEntity = new TableRelationshipEntity();
                    relationshipEntity.setRelationFieldName(columnName);
                    relationshipEntity.setRelationFieldType(typeName);
                    relationshipEntity.setRelationFieldDesc(remarks);
                    relationshipEntity.setRelationTableName(tableName);
                    relationshipEntity.setRelationTableDesc(null);
                    relationshipEntity.setRelationDbname(req.getDbName());
                    relationshipEntity.setRelationSourceId(req.getSourceId());
                    relationshipEntity.setRelationSourceName(confConnect.getName());
                    shipList.add(relationshipEntity);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        } finally {
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException e) {
                log.error(e.getMessage());
            }
        }
        return shipList;
    }

    private Boolean getSourceColumns(List<TableRelationshipEntity> shipList, TableInfoFieldRes tableInfoField) {
        if(CollectionUtil.isEmpty(shipList))
            return false;

        List<TableRelationshipEntity> collect = shipList.stream().filter(
                s -> Objects.equals(s.getRelationCollTableId(),tableInfoField.getId())
                        && s.getRelationTableName().equalsIgnoreCase(tableInfoField.getTableName()))
                .collect(Collectors.toList());
        if(CollectionUtil.isEmpty(collect))
            return false;

        collect.stream().forEach(t -> {
            t.setRelationTableDesc(tableInfoField.getDesc());
        });
        tableInfoField.setRelationshipes(collect);

        return true;
    }

    /***
     *  接口参数修改添加
     * @param confConnect
     * @return
     */
    public ConfConnect testJiekouParamsUpdate(ConfConnect confConnect) {
        if (confConnect.getInterfaceUrl() == null || confConnect.getFormat() == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        List<ConnectFieldInfo> connectFieldInfos = confConnect.getConnectFieldInfos();
        if(connectFieldInfos==null||connectFieldInfos.size()==0){
            return confConnect;
        }
        Map<String, ConnectFieldInfo> collect = connectFieldInfos.stream().collect(Collectors.toMap(ConnectFieldInfo::getFieldName, Function.identity(), (key1, key2) -> key2));
        /**处理动态{{}}引用问题*/
        for(Map.Entry<String, ConnectFieldInfo> entry : collect.entrySet()){
            String mapKey = entry.getKey();
            ConnectFieldInfo mapValue = entry.getValue();
            /**判断是否可解析*/
            if(getParamsBoolean(mapValue.getFieldValue())){
                  /**动态参数引用当中的值*/
                if(confConnect.getVariableName().equals(getParams(mapValue.getFieldValue()))){
                    if(confConnect.getTokenUrl()!=null){
                         /**临时存放之前的值*/
                        mapValue.setPastValue("{{"+getParams(mapValue.getFieldValue())+"}}");
                         /**存放token*/
                        mapValue.setFieldValue(getHttpToken(confConnect.getTokenUrl()));
                        collect.put(mapKey,mapValue);
                    }
                    if(confConnect.getVariableValue()!=null){
                        /**解析data*/
                        String params = getParams(confConnect.getVariableValue());
                        /**临时存放之前的值*/
                        mapValue.setPastValue(confConnect.getVariableValue());
                        /**解析日期*/
                        String data = detailTime(params,1);
                        mapValue.setFieldValue(data);
                        collect.put(mapKey,mapValue);
                    }
                }
            }
        }
        /**处理type=2 拼接url问题*/
        for(Map.Entry<String, ConnectFieldInfo> entry : collect.entrySet()){
            String mapKey = entry.getKey();
            if(entry.getValue().getFieldType()==2){
                Map<String, Object> map=new HashMap();
                map.put(mapKey,getParams(entry.getValue().getFieldValue()));
                String url = confConnect.getTmeInterfaceUrl();
                if(StringUtils.isEmpty(url)){
                    url = confConnect.getInterfaceUrl();
                }
                confConnect.setTmeInterfaceUrl( appendUrl(url,map));
            }
        }
        return confConnect;
    }

    /***
     * 更新解析当前日期
     */
//    @Scheduled(cron = "0 */10 * * * ?")
    @Override
    public void cronTimeDate() {
        List<ConnectFieldInfo> list = connectFieldInfoService.list(new QueryWrapper<ConnectFieldInfo>().eq("field_type", 3).eq("field_alias", "$variable^"));
        if(list!=null&&list.size()>0){
            for(ConnectFieldInfo connectFieldInfo:list){
                /***解析时间获取现在时间*/
                String params = getParams(connectFieldInfo.getFieldValue());
                String data = detailTime(params,2);
                String timedata = detailTime(params,1);
                connectFieldInfo.setFieldValue(data);
                /**需要更新的字段*/
                List<ConnectFieldInfo> list1 = connectFieldInfoService.list(new QueryWrapper<ConnectFieldInfo>().eq("connect_id", connectFieldInfo.getConnectId()).notIn("field_alias", "$variable^"));
                if(list1!=null){
                    for(ConnectFieldInfo connectFieldInfo1:list1){
                        /**判断是否可以解析时间*/
                        if(isDate(connectFieldInfo1.getFieldValue(),data)){
                            /**更新时间*/
                            connectFieldInfo1.setFieldValue(timedata);
                            connectFieldInfoService.updateById(connectFieldInfo1);
                        }
                    }
                }
            }
        }
    }

    @Override
    public void enableOrDisable(Long id) {
        ConfConnect confConnect = baseMapper.selectById(id);
        if(confConnect == null)
            throw new RRException("没有该数据源");

        ConfConnect updateConnect = new ConfConnect();
        updateConnect.setId(confConnect.getId());
        if(confConnect.getConnectEnable() == null)
            confConnect.setConnectEnable(1);
        int connectEnable = confConnect.getConnectEnable().intValue() == 0 ? 1 : 0;//是否启用 0:不启用 1:启
        updateConnect.setConnectEnable(connectEnable);
        baseMapper.updateById(updateConnect);
    }

    /**
     * 在指定url后追加参数
     * @param url
     * @param data 参数集合 key = value
     * @return
     */
    private  String appendUrl(String url, Map<String, Object> data) {
        String newUrl = url;
        StringBuffer param = new StringBuffer();
        for (String key : data.keySet()) {
            param.append(key + "=" + data.get(key).toString() + "&");
        }
        String paramStr = param.toString();
        paramStr = paramStr.substring(0, paramStr.length() - 1);
        if (newUrl.indexOf("?") >= 0) {
            newUrl += "&" + paramStr;
        } else {
            newUrl += "?" + paramStr;
        }
        return newUrl;
    }

    /**
     * 解析提取数据 {{}} 和 ${} 数据
     * @param fieldValue
     * @return
     */
    public String getParams(String fieldValue){
        String reg = "\\{\\{(.+?)\\}\\}";
        List<String> params = getParams(reg, fieldValue);
        if(params!=null&&params.size()>0){
            return params.get(0);
        }else{
            String regs = "\\$\\{([^}]*)\\}";
            List<String> data = getParams(regs, fieldValue);
            if(data!=null&&data.size()>0){
                return data.get(0);
            }
        }
        return fieldValue;
    }

    /***
     * 判断 {{}} 是否可解析
     * @param fieldValue
     * @return
     */
    public boolean getParamsBoolean(String fieldValue){
        String reg = "\\{\\{(.+?)\\}\\}";
        List<String> params = getParams(reg, fieldValue);
        if(params!=null&&params.size()>0){
            return true;
        }
        return  false;
    }

    /**
     * 根据正则表达式获取文本中的变量名列表
     * @param pattern
     * @param content
     * @return
     */
    public List<String> getParams(String pattern, String content) {
        Pattern p = Pattern.compile(pattern);
        Matcher m = p.matcher(content);
        List<String> result = new ArrayList<String>();
        while (m.find()) {
            result.add(m.group(1));
        }
        return result;
    }

    /**
     * 时间解析 忽越大小写
     * @param contentInfo  ${value}中的值
     * @param type         类型 1解析并计算时间 ,否则返回表达式
     * @return
     */
    public String detailTime(String contentInfo,int type){
        /**判断字符串是否存在*/
        if (contentInfo.indexOf("Y")!=-1){
            contentInfo=contentInfo.replace("Y","y");
        }
        if (contentInfo.indexOf("m")!=-1){
            contentInfo=contentInfo.replace("m","M");
        }
        if (contentInfo.indexOf("D")!=-1){
            contentInfo=contentInfo.replace("D","d");
        }
        Pattern number = Pattern.compile("[^0-9]");
        Matcher matchers = number.matcher(contentInfo);
        String alls = matchers.replaceAll("");
        /**判断是否需要计算查找是否有数字存在*/
        if(StringUtils.isNotBlank(alls)){
            String substring = contentInfo.substring(0, contentInfo.length() - 2);
            String substring1 = contentInfo.substring(contentInfo.length() - 2, contentInfo.length());
            if(type==1){
                return getDate(Integer.parseInt(substring1), substring);
            }else{
                return substring;
            }
        }else{
            if(type==1){
                return getDate(0, contentInfo);
            }else{
                return contentInfo;
            }
        }
    }

    /***
     * 时间计算
     * @param i    当前天  +1 为当前天加上一天
     * @param substring  日期格式  yyyymmdd
     * @return
     */
    public  String getDate(Integer i,String substring){
        Calendar cal = Calendar.getInstance();
        cal.add(Calendar.DATE, i);
        Date d = cal.getTime();
        SimpleDateFormat sp = new SimpleDateFormat(substring);
        String yesterday = sp.format(d);// 获取昨天日期
        return yesterday;
    }

    /***
     * 判断是否可以解析日期
     * @param str     日期     20001122
     * @param dataStr 日期格式  yyyymmdd
     * @return
     */
    public boolean isDate(String str,String dataStr) {
        boolean result = true;
        SimpleDateFormat format = new SimpleDateFormat(dataStr);
        try {
            format.parse(str);
        }catch (Exception e){
            result = false;
        }
        return result;
    }
    /***
     * 请求第三方获取token秘钥
     * @param url
     * @return
     */
    public String getHttpToken(String url){
        if (!checkHttp(url)) {
            return url;
        }
        String body = HttpRequest.get(url).execute().body();
        return body;
    }


    /**
     * @Description 根据数据源ID，修改数据源关联的数据库名
     * @Author  chengweiping
     * @Date   2021/1/8 15:19
     */
    @Override
    public void updateCollectionTaskDbName(Integer  sourceId,String dbName){
        try{
            //有源和有库则修改关联的采集任务对应的数据库
            if(sourceId!=null&&sourceId>0&&StringUtils.isNotBlank(dbName)){
                CollectionTaskUpdateDbDto collectionTaskUpdateDbDto=new CollectionTaskUpdateDbDto();
                collectionTaskUpdateDbDto.setDbName(dbName);
                collectionTaskUpdateDbDto.setSourceId(sourceId);
                confConnectMapper.updateCollectionTaskDbName(collectionTaskUpdateDbDto);
            }

        }catch (Exception e){
            e.getMessage();
        }
    }

    //获取关联数据源创建人
    public List<UserCreaterVO>  selectCreaterList(){
        return confConnectMapper.selectCreaterList();
    }
}
